package model

import (
	"census/common/utils"
	"census/context"
	"fmt"
	"github.com/golib/logger"
)

//调度者

type Processor struct {
	appName string
	msgChan chan *context.AppRequest
	sem     chan struct{}
}

func NewProcessor(appName string, parallelMaxMsg int) *Processor {
	var p = new(Processor)
	p.appName = appName
	// message
	p.msgChan = make(chan *context.AppRequest)
	//max
	if parallelMaxMsg <= 0 {
		parallelMaxMsg = 1
	}
	p.sem = make(chan struct{}, parallelMaxMsg)
	return p
}

func (p *Processor) Write(msg *context.AppRequest) (err error) {
	ch, err := GetChannel(p.GetAppName())
	if err != nil {
		return
	}
	select {
	case <-ch.ProcessorStop:
		err = errProcessorHasStop
	case p.msgChan <- msg:
	}
	return
}

func (p *Processor) GetAppName() string {
	return p.appName
}

func (p *Processor) Valid() bool {
	ch, err := GetChannel(p.GetAppName())
	if err != nil {
		return false
	}
	select {
	case <-ch.ProcessorStop:
		return false
	default:
		return true
	}
}

func (p *Processor) Do() {
	ch, err := GetChannel(p.GetAppName())
	if err != nil {
		return
	}
	for {
		//process
		p.sem <- struct{}{}
		select {
		case item := <-p.msgChan:
			go p.do(item)
		case <-ch.ProcessorStop:
			return
		}
	}
}

func (p *Processor) do(item *context.AppRequest) {
	defer func() {
		//when pool use func processor.GetSem(), here may be blocked
		//so select default
		select {
		case <-p.sem:
		default:
		}
		if err := recover(); err != nil {
			logger.Warning("processor-error", utils.MakeUUID(), "appName:["+item.AppName+"]", fmt.Sprintf("error msg:%v", err))
		}
	}()
	var taskConf = context.GetAppTask(item.AppName, item.Action)
	if taskConf == nil {
		logger.Warning("task-init", utils.MakeUUID(), "appName:["+item.AppName+"]", fmt.Sprintf("task conf not exist, taskId:%v", item.Action))
		return
	}
	//创建task
	var task = NewTask(item.AppName, item.Action, item.Data)
	//注册回调
	task.On(TaskStatusReady, item.GetEvent(context.EventTaskReady))
	task.On(TaskStatusBuffer, item.GetEvent(context.EventTaskBuffer))
	task.On(TaskStatusStart, item.GetEvent(context.EventTaskStart))
	task.On(TaskStatusSuccess, item.GetEvent(context.EventTaskSuccess))
	task.On(TaskStatusFail, item.GetEvent(context.EventTaskFail))
	//执行ready
	task.Emit(TaskStatusReady)
	//是否进入buffer
	pool, err := GetPool(p.GetAppName())
	if err != nil {
		logger.Warning("task-init", utils.MakeUUID(), "appName:["+item.AppName+"]", err.Error())
		return
	}
	logger.Info("task-ready", task.GetTaskUUId(), "appName:["+item.AppName+"]", "task-data", item.Data)
	pool.DispatchTask(task)
}

func (p *Processor) Stop() {
	ch, err := GetChannel(p.GetAppName())
	if err == nil {
		close(ch.ProcessorStop)
		//等待正在进行的任务处理结束
		<-ch.AllTaskStop
	}
}

func (p *Processor) GetSem() <-chan struct{} {
	return p.sem
}
